package operator;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * min 算子 分组后 统计组内最小值 如果有多余字段 任意组装
 * minBy 算子 分组 统计组内 最小值 如果有多余字段 按照第一次出现的进行拼接 也可以指定first为false
 */
public class MinMinByDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        // 武汉市,江夏区,1000
        // 武汉市,洪山区,800
        // 武汉市,硚口区,500
        // 武汉市,东西湖区,2000
        // 十堰市,张湾区,1000
        // 十堰市,十八区,500
        DataStreamSource<String> data = env.socketTextStream("localhost", 8888);

        SingleOutputStreamOperator<Tuple3<String, String, Integer>> tuples = data.map(line -> {
            String[] fields = line.split(",");
            return Tuple3.of(fields[0],fields[1],Integer.parseInt(fields[2]));
        }).returns(Types.TUPLE(Types.STRING,Types.STRING, Types.INT));

        KeyedStream<Tuple3<String,String, Integer>, String> keyed = tuples.keyBy(t -> t.f0);

        // 调用min算子
        // SingleOutputStreamOperator<Tuple3<String,String, Integer>> min = keyed.min(2);

        // 调用minBy算子
        SingleOutputStreamOperator<Tuple3<String, String, Integer>> min = keyed.minBy(2, false);

        min.print();

        env.execute("min operator");
    }
}
